package dim

import modes.AreaDetail
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

object AreaDetailInfo {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val data: DataStream[(String, String)] = env.addSource(new AreaHBaseReader)

    val areaDetail: DataStream[AreaDetail] = data.map(row => {
      val rowKey: Int = row._1.toInt
      val items: Array[String] = row._2.split("-")
      AreaDetail(rowKey, items(5), items(6).toInt)
    })

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    tEnv.createTemporaryView("lagou_area", areaDetail)

    val sql: String =
      """
        |select a.id areaid,a.name aname, a.pid cid,b.name city, c.id proid,c.name province
        |from lagou_area a
        |inner join lagou_area b on a.pid = b.id
        |inner join lagou_area c on b.pid = c.id
        |""".stripMargin

    val result: Table = tEnv.sqlQuery(sql)

    val resultStream: DataStream[String] = tEnv.toRetractStream[Row](result).map(x => {
      val row: Row = x._2
      val areaId: String = row.getField(0).toString
      val aname: String = row.getField(1).toString
      val cid: String = row.getField(2).toString
      val city: String = row.getField(3).toString
      val proid: String = row.getField(4).toString
      val province: String = row.getField(5).toString
      areaId + "," + aname + "," + cid + "," + city + "," + proid + "," + province
    })

    resultStream.addSink(new AreaHBaseSink)

    env.execute()
  }
}
